package cn.doitedu.rtdw.etl;

import cn.doitedu.rtdw.udfs.GetNullUdf;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/12/18
 * @Desc: 学大数据，上多易教育
 *   推荐位展示点击olap分析基础表
 **/
public class Job06_recommend_olap_base {
    public static void main(String[] args) {

        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 创建kafka中dwd行为明细topic的映射表
        tenv.executeSql(
                " create table dwd_events_kafka (                                 "+
                        "      session_id  string                                 "+
                        "     ,user_name    string                                "+
                        "     ,event_id     string                                 "+
                        "     ,action_time   bigint                               "+
                        "     ,properties  map<string,string>                     "+
                        "     ,user_id  bigint                                    "+
                        "     ,create_time timestamp(3)                           "+
                        "     ,modify_time timestamp(3)                           "+
                        "     ,province string                                    "+
                        "     ,city string                                        "+
                        "     ,region string                                      "+
                        "     ,page_type string                                   "+
                        "     ,page_service string                                "+
                        "     ,device_type string                                 "+
                        "     ,release_channel string                             "+
                        "     ,rt as to_timestamp_ltz(action_time,3)              "+
                        "     ,watermark for rt as rt - interval '0' second       "+
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'dwd-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    " +
                        "     'properties.group.id' = 'doit43-2',                 " +
                        "     'scan.startup.mode' = 'latest-offset',              " +
                        "     'value.format'='json',                              " +
                        "     'value.json.fail-on-missing-field'='false',         " +
                        "     'value.fields-include' = 'EXCEPT_KEY'   )           "
        );


        tenv.createTemporaryFunction("get_null", GetNullUdf.class);
        tenv.executeSql(
                " with tmp as (                                                                             "+
                        "     select                                                                                "+
                        "         user_id,                                                                          "+
                        "     	event_id,                                                                           "+
                        "     	action_time,                                                                        "+
                        "     	properties['rec_region_id'] as rec_region_id,                                       "+
                        "     	properties['rec_tracking_id'] as rec_tracking_id,                                   "+
                        "     	rt	                                                                                "+
                        "     from dwd_events_kafka                                                                 "+
                        "     where event_id in ('rec_show','rec_click')                                            "+
                        " )                                                                                         "+
                        "                                                                                           "+
                        "                                                                                           "+
                        " select                                                                                    "+
                        "   window_start,                                                                         "+
                        " 	window_end,                                                                             "+
                        " 	user_id,                                                                                "+
                        " 	rec_region_id,                                                                          "+
                        " 	count(1) filter(where event_id = 'rec_show' ) as show_cnt,                              "+
                        " 	count(1) filter(where event_id = 'rec_click' ) as click_cnt,                            "+
                        " 	if(count(1) filter(where event_id = 'rec_click' ) > 0 ,user_id, get_null() ) as click_user_id   "+
                        " from table(                                                                               "+
                        "     tumble(table tmp,descriptor(rt),interval '1' minute)                                  "+
                        " )                                                                                         "+
                        "                                                                                           "+
                        " group by                                                                                  "+
                        "     window_start,                                                                         "+
                        " 	window_end,                                                                             "+
                        " 	user_id,                                                                                "+
                        " 	rec_region_id                                                                           "
        );





    }

}
